package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
public class HotArticleKafkaStream {

    @Bean
    public KStream<String,String> process(StreamsBuilder streamsBuilder){

        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);  //{article:111 type:likes add:1},{article:111 type:likes add:1},{article:111 type:views add:1}

        stream
                .map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(String key, String value) {

                        UpdateArticleMess articleMess = JSON.parseObject(value, UpdateArticleMess.class);

                        return new KeyValue<>(articleMess.getArticleId()+"",articleMess.getType()+":"+articleMess.getAdd());   // key:111 value: likes:1 , key:111 value: likes:1 key:111 value: views:1
                    }
                })
                .groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .aggregate(
                        new Initializer<String>() {
                            @Override
                            public String apply() {
                                return "LIKES:0,VIEWS:0,COLLECTION:0,COMMENT:0";
                            }
                        },
                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String init) {   //value: likes:1    init:""LIKES:2,VIEWS:,COLLECTION:0,COMMENT:0""
                                if(StringUtils.isBlank(value)){
                                    return init;      //"LIKES:0,VIEWS:0,COLLECTION:0,COMMENT:0"
                                }

                                int likes=0,views=0,collection=0,comment=0;

                                String[] split1 = init.split(",");    //split1 ["likes:0","views:1",....]
                                for (String s : split1) {
                                    String[] split2 = s.split(":");
                                    switch (split2[0]){
                                        case "LIKES":
                                            likes=Integer.parseInt(split2[1]);
                                            break;
                                        case "VIEWS":
                                            views=Integer.parseInt(split2[1]);
                                            break;
                                        case "COLLECTION":
                                            collection=Integer.parseInt(split2[1]);
                                            break;
                                        case "COMMENT":
                                            comment=Integer.parseInt(split2[1]);
                                            break;
                                    }
                                }


                                String[] split = value.split(":");   //split[0]="views" split[1]=1
                                switch (split[0]){
                                    case "COLLECTION":
                                        collection+=Integer.parseInt(split[1]);
                                        break;
                                    case "COMMENT":
                                        comment+=Integer.parseInt(split[1]);
                                        break;
                                    case "LIKES":
                                        likes+=Integer.parseInt(split[1]);
                                        break;
                                    case "VIEWS":
                                        views+=Integer.parseInt(split[1]);
                                        break;
                                }

                                return String.format("LIKES:%d,VIEWS:%d,COLLECTION:%d,COMMENT:%d",likes,views,collection,comment);   //"LIKES:0,VIEWS:2,COLLECTION:0,COMMENT:0"
                            }
                        },
                        Named.as("hotraticle_stream")
                )
                //"LIKES:2,VIEWS:1,COLLECTION:0,COMMENT:0"

                .toStream()
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        return new KeyValue<>(key.key().toString(),formatData(key.key().toString(),value));  //110, "LIKES:2,VIEWS:1,COLLECTION:0,COMMENT:0"
                    }
                })//{articleId:0 ,view:1 , collect: 0,comment: 0,like:2}
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;
    }



    private String formatData(String key,String value){    //110, "LIKES:2,VIEWS:1,COLLECTION:0,COMMENT:0"
        ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();  //{articleId:0 ,view:1 , collect: 0,comment: 0,like:2}
        articleVisitStreamMess.setArticleId(Long.valueOf(key));

        String[] split = value.split(",");
        for (String s : split) {
            String[] split1 = s.split(":");
            switch (split1[0]){
                case "LIKES":
                    articleVisitStreamMess.setLike(Integer.parseInt(split1[1]));
                    break;
                case "VIEWS":
                    articleVisitStreamMess.setView(Integer.parseInt(split1[1]));
                    break;
                case "COLLECTION":
                    articleVisitStreamMess.setCollect(Integer.parseInt(split1[1]));
                    break;
                case "COMMENT":
                    articleVisitStreamMess.setComment(Integer.parseInt(split1[1]));
                    break;
            }
        }

        return JSON.toJSONString(articleVisitStreamMess);

    }
}
